package com.google.code.esper.spring;

import com.espertech.esper.client.*;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.config.AbstractFactoryBean;

/**
 *
 */
public class EPServiceProviderFactoryBean extends AbstractFactoryBean<EPServiceProvider> implements BeanNameAware, FactoryBean<EPServiceProvider>, InitializingBean, DisposableBean {

    private String name;
    private Configuration configuration;
    private List<? extends Object> listeners = Collections.emptyList();
    private UnmatchedListener unmatchedListener;
    private Map<String, ? extends Object> variables;
    private List<? extends EPServiceProviderPostprocessor> postprocessors = Collections.emptyList();

    @Override
    public Class<? extends EPServiceProvider> getObjectType() {
        return EPServiceProvider.class;
    }

    private EPServiceProvider getProvider() {
        if (configuration == null)
            return EPServiceProviderManager.getProvider(name);
        else
            return EPServiceProviderManager.getProvider(name, configuration);
    }

    @Override
    protected EPServiceProvider createInstance() throws Exception {
        EPServiceProvider provider = getProvider();

        // register listeners
        for (Object listener : listeners) {
            if (listener instanceof EPServiceStateListener) {
                provider.addServiceStateListener((EPServiceStateListener) listener);
            } else if (listener instanceof EPStatementStateListener) {
                provider.addStatementStateListener((EPStatementStateListener) listener);
            } else {
                throw new IllegalArgumentException("Listeners of type " + listener.getClass().getSimpleName() + " are not supported");
            }
        }

        if (unmatchedListener != null) {
            provider.getEPRuntime().setUnmatchedListener(unmatchedListener);
        }

        // postprocess provider
        for (EPServiceProviderPostprocessor postprocessor : postprocessors) {
            postprocessor.postprocess(provider);
        }

        // register statements
        EPAdministrator administrator = provider.getEPAdministrator();
        BeanFactory beanFactory = getBeanFactory();
        if (beanFactory instanceof ListableBeanFactory) {
            ListableBeanFactory listableBeanFactory = (ListableBeanFactory)beanFactory;
            Collection<EPStatementRegistrar> registrars = BeanFactoryUtils.beansOfTypeIncludingAncestors(listableBeanFactory, EPStatementRegistrar.class, true, true).values();
            for (EPStatementRegistrar registrar : registrars) {
                registrar.registerStatements(administrator);
            }
        }

        return provider;
    }

    @Override
    protected void destroyInstance(EPServiceProvider instance) throws Exception {
        instance.destroy();
    }

    public void setBeanName(String name) {
        this.name = name;
    }

    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    public void setListeners(List<? extends Object> listeners) {
        this.listeners = listeners;
    }

    public void setUnmatchedListener(UnmatchedListener unmatchedListener) {
        this.unmatchedListener = unmatchedListener;
    }

    public void setPostprocessors(List<? extends EPServiceProviderPostprocessor> postprocessors) {
        this.postprocessors = postprocessors;
    }
}
